package com.o2o.cleaning.month.platform.ebusiness_plat.meituan

import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}
import org.apache.spark.sql.functions.lit

object Address_Gov_Store_2019_Meituan {

  def meituan_handleAddress(spark: SparkSession, platform: String): Unit = {


    /**
      * 先： addressdata   后：sourcedata
      */
    //      val neeedNo = "addressdata"
    val neeedNo = "addressdata"

    //采集详情里的地址数据路径
    val sourcedata_path = Meituan_config.addr_path_from_detail

    // mongoDB 地址新采集源数据
    val mongodbData_new_path = Meituan_config.addressPath_from_mongo

    // 上月处理后的地址数据保存路径
    val oldAddr_OBS_path = Meituan_config.resultAddressForRead

    //截至本月全量地址保存路径
    val resultSave2OBS = Meituan_config.resultAddress
    //建林维护地址映射表
    val AddressMapping = Meituan_config.AddressMapping

    //======================================= 读取数据 ==================================================================

    /**
      * 一、0. 读取地址映射表
      */
    //地址映射表
    val standardTableDF = Relation_Address_meituan.getAddressMapping(spark, AddressMapping)

    /**
      * 先： addressdata   后：sourcedata
      */
    if (neeedNo == "addressdata") {

      /**
        * 一、1. 读取新增数据
        */
      // 读取平台新增数据
    /*  val mongodbData_newdf: DataFrame = spark.read.json(mongodbData_new_path)
        .filter("regional_ID !='-1' and regional_ID !='0'")
        .select("shopId", "address", "latitude", "longitude", "regional_ID", "province")

      /** 一、1.1 更改一些特殊的 regional_ID */
      val mongodbData_newDF = Relation_Address_meituan.updateOldRegionalID(spark, mongodbData_newdf)
        .drop("province")*/

      // 查找 不能关联平台新增地址数据的regional_ID 省级ID忽略不管 有则需要添加 转换region_id为地址表里的region_id （Relation_Address.updateOldRegionalID）
      //      println("addrNoJoin : 有则需要添加 转换region_id为地址表里的region_id ")
      //      mongodbData_newDF.select("regional_ID").dropDuplicates().except(standardTableDF.selectExpr("region_id_nosplit as regional_ID")).show(50)

      //美团外卖的地址集合未更新，直接读取上个月的结果集合

      val mongodbData_newDF: DataFrame = spark.read.json(oldAddr_OBS_path + "/mongoDB_address")
        .filter("regional_ID !='-1' and regional_ID !='0'")
        .select("shopId", "address", "latitude", "longitude", "regional_ID", "province")


      /**
        * 一、2. 读取旧数据
        */
      // 读取旧的地址数据    === mongoDB_address 为处理的采集地址数据
      val oldDF: DataFrame = spark.read.json(oldAddr_OBS_path + "/mongoDB_address")
        .select("shopId", "address", "latitude", "longitude", "regional_ID")


      /**
        * 一、3. 合并新旧数据     :看自己平台具体数据，覆盖或保持旧数据
        */
      // 若新旧数据有交集，用新数据覆盖 （看自己平台具体数据，覆盖或保持旧数据）
      val oldDF_noNew: DataFrame = oldDF.join(mongodbData_newDF.selectExpr("shopId", "regional_ID as regional_ID_new"), Seq("shopId"), "left")
        .filter("regional_ID_new is null").drop("regional_ID_new")

      val needDF: Dataset[Row] = mongodbData_newDF.select("shopId", "address", "latitude", "longitude", "regional_ID")
        .union(oldDF_noNew.select("shopId", "address", "latitude", "longitude", "regional_ID"))

      //  若没有新增 直接读旧数据 注释掉 读取新增数据 和 合并新旧数据 的代码(这里的新增指的是佳迪采集的数据，是否采集到的地址数据)
      //        val needDF: DataFrame = spark.read.json(oldAddr_OBS_path).select("shopId", "address", "latitude", "longitude", "regional_ID")


      //======================================= 关联 标准地址映射表 ========================================================


      /**
        * 二、地址数据 关联 标准地址表
        *
        */
      // 修改经纬度和关联标准地址数据
      val data_mapping = Relation_Address_meituan.withAddressMapping(spark, needDF, standardTableDF)
        .drop("uniqueid_nosplit").withColumnRenamed("uniqueid", "regional_ID")
        .filter("regional_ID !='0'")
      //    println("校验地址表中的数据是否有数据没有关联上：")
      //    println(data_mapping.selectExpr("shopId","regional_ID").where("regional_ID ='0'").count())
      //    println("data_mapping:" + data_mapping.count())


      //======================================= 处理特殊字段 ==============================================================

      /**
        * 匹配 开发区
        * NEED : city,district,address,registration_institution   无字段先打上'-1'
        */
      val addr_aedzId_df = Relation_Address_meituan.match_AedzId(spark, data_mapping.withColumn("registration_institution", lit("-1")))


      /**
        * 匹配： 中山市、东莞市的镇
        */
      val zs_dg_town_result2: DataFrame = Relation_Address_meituan.match_ZSDG(spark, platform, addr_aedzId_df)

      //匹配中山东莞的id,并将中山东莞的town的值打到district中
      val math_DF: DataFrame = Relation_Address_meituan.match_ZSDG_ID(spark, zs_dg_town_result2)

      val resultDataTable = math_DF

      //保存
      resultDataTable.repartition(1).write.json(resultSave2OBS + "/mongoDB_address")


    }


    /**
      * 先： addressdata   后：sourcedata
      */
    if (neeedNo == "sourcedata" || neeedNo == "addressdata") {

      // 详情数据地址
      val sourcedataDF: DataFrame = spark.read.json(sourcedata_path).select("shopId", "address", "latitude", "longitude", "regional_ID", "province")

      //处理完的：截止本月的采集地址
      val address_mongodb_DF: DataFrame = spark.read.json(resultSave2OBS + "/mongoDB_address").filter("regional_ID!='0'")

      // 详情数据地址 （不包括采集地址）
      val sourcedataDF_New: DataFrame = sourcedataDF.select("shopId").except(address_mongodb_DF.select("shopId"))
        .join(sourcedataDF, Seq("shopId"), "left")

      // 转换旧的 region_id为地址表里的region_id
      val needDF = Relation_Address_meituan.updateOldRegionalID(spark, sourcedataDF_New).drop("province")
      // 查找 不能关联平台新增地址数据的regional_ID 省级ID忽略不管 有则需要添加 转换region_id为地址表里的region_id （Relation_Address.updateOldRegionalID）
      println("addrNoJoin : 有则需要添加 转换region_id为地址表里的region_id ")
      //        needDF.select("regional_ID").dropDuplicates().except(standardTableDF.selectExpr("region_id_nosplit as regional_ID")).show(50)


      //======================================= 关联 标准地址映射表 ========================================================


      /**
        * 二、地址数据 关联 标准地址表
        *
        */
      // 修改经纬度和关联标准地址数据
      val data_mapping = Relation_Address_meituan.withAddressMapping(spark, needDF, standardTableDF).drop("uniqueid_nosplit").withColumnRenamed("uniqueid", "regional_ID")
        .filter("regional_ID !='0'")
      //    println("校验地址表中的数据是否有数据没有关联上：")
      //    println(data_mapping.selectExpr("shopId","regional_ID").where("regional_ID ='0'").count())
      //    println("data_mapping:" + data_mapping.count())


      //======================================= 处理特殊字段 ==============================================================

      /**
        * 匹配 开发区
        * NEED : city,district,address,registration_institution   无字段先打上'-1'
        */
      val addr_aedzId_df = Relation_Address_meituan.match_AedzId(spark, data_mapping.withColumn("registration_institution", lit("-1"))).drop("registration_institution")


      /**
        * 匹配： 中山市、东莞市的镇
        */
      val zs_dg_town_result2: DataFrame = Relation_Address_meituan.match_ZSDG(spark, platform, addr_aedzId_df)

      //匹配中山东莞的id,并将中山东莞的town的值打到district中
      val resultDataTable: DataFrame = Relation_Address_meituan.match_ZSDG_ID(spark, zs_dg_town_result2)

      resultDataTable.repartition(1).write
        .json(resultSave2OBS + "/sourceData_NewAdd")

    }

  }


}
